﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace Dhgms.TplHelper.Controller.JobSequence
{
    using System.Collections.Concurrent;
    using System.Threading.Tasks;

    public abstract class ParallelProducerToSerialConsumerSequence<TSerialDataClass, TParallelProducerDataType>
        where TSerialDataClass : class
        where TParallelProducerDataType : class
    {
        private readonly BlockingCollection<TSerialDataClass> serialDataCollection;

        protected ParallelProducerToSerialConsumerSequence()
        {
            this.serialDataCollection = new BlockingCollection<TSerialDataClass>();
        }

        private void ProcessBlockingCollection()
        {
            try
            {
                // Consume consume the BlockingCollection 
                while (true)
                {
                    // this call blocks until there is something to process
                    // or until CompleteAdding has been signalled
                    this.DoSerialActionOnItem(this.serialDataCollection.Take());
                }
            }
            catch (InvalidOperationException)
            {
                // An InvalidOperationException means that Take() was called on a completed collection
            }
        }

        protected abstract void DoSerialActionOnItem(TSerialDataClass take);

        public void Execute()
        {
            var serialTask = System.Threading.Tasks.Task.Factory.StartNew(ProcessBlockingCollection);

            var parallelProducerData = new List<TParallelProducerDataType>();

            if (parallelProducerData != null && parallelProducerData.Count > 0)
            {
                Parallel.ForEach(
                    parallelProducerData,
                    item => DoParallelConsumption(item));
            }

            serialDataCollection.CompleteAdding();
            serialTask.Wait();
        }

        protected TSerialDataClass DoParallelConsumption(TParallelProducerDataType parallelProducerItem)
        {
            var serialItem = GetSerialItemFromParallelItem(parallelProducerItem);
            if (serialItem != null)
            {
                serialDataCollection.Add(serialItem);
            }
        }

        protected abstract TSerialDataClass GetSerialItemFromParallelItem(TParallelProducerDataType parallelProducerItem);
    }
}
